package com.itheima.kafka.boot.listener;

import com.alibaba.fastjson.JSON;
import com.itheima.kafka.boot.controller.HelloController;
import com.itheima.kafka.boot.pojo.User;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

/**
 * @Description: 监听消息
 * @Version: V1.0
 */
@Component
public class HelloListener {

/*

    @KafkaListener(topics = HelloController.TOPIC)
    public void receiverMessage(ConsumerRecord<?, ?> record) {

//        if (record != null) {
//        }
        // 判断非空的标准
        Optional<? extends ConsumerRecord<?, ?>> optional = Optional.ofNullable(record);
        if (optional.isPresent()) {

            record = optional.get();

            System.out.println(record.key());
            System.out.println(record.value());
        }

    }
*/

    @KafkaListener(topics = HelloController.TOPIC)
    public void receiverMessage(String message) {
        if (!StringUtils.isEmpty(message)) {
            System.out.println(message);

            User user = JSON.parseObject(message, User.class);
            System.out.println(user.getAge());
            System.out.println(user.getUsername());
        }
    }
}
